package com.ustcinfo.study.hbase.xngroup.mouyiquan;

import com.ustcinfo.ishare.bdp.hbase.HBaseOperationDao;
import org.apache.hadoop.hbase.client.Put;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;


/**
 * @Author: Yiquan Mou
 * @Email: ctbumyq@gmail.com
 * @Description:   数据迁移类 从Mysql数据库迁移到HBase
 * @Date: Created in 19:13 2018/8/1
 */
public class DataMigration {

    public static void startMvDate() throws SQLException,ClassNotFoundException{
        //我们建立三个变量
        Connection con = null; //用于连接数据库
        Statement str = null;//用于向数据库发送sql语句
        ResultSet resu = null;//用于处理我们查找到的结果集


        con = JdbcUtils.GetConnection();
        //创建sql语句
        String sql = "select * from transfer_task_to";
        //获取发送sql语句Statement对象
        str = (Statement) con.createStatement();
        //执行sql
        resu = str.executeQuery(sql);
        List<TransferTaskToEntity> list = new ArrayList<>();
        //打印出我们查询到的数据
        while (resu.next()) {
            TransferTaskToEntity entity = new TransferTaskToEntity();
            //设置写到 F1的数据
            entity.setId(resu.getInt("id"));
            entity.setTask_id(resu.getString("task_id"));
            entity.setCreate_time(resu.getDate("create_time"));
            entity.setStart_time(resu.getDate("start_time"));
            entity.setOut_filter_number(resu.getString("out_filter_number"));
            entity.setFrom_run_param(resu.getString("from_run_param"));

            //设置写到 F2的数据

            entity.setCurrent_retry_times(resu.getString("current_retry_times"));
            entity.setRetry_flag(resu.getString("retry_flag"));
            entity.setParent_task_id(resu.getString("parent_task_id"));
            entity.setEnd_time(resu.getDate("end_time"));
            entity.setKill_time(resu.getDate("kill_time"));
            entity.setFailure_time(resu.getDate("failure_time"));

            list.add(entity);
        }
        con.close();

        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        List<Put> puts = new ArrayList<>();
        for(int i=0;i<list.size();i++){

            TransferTaskToEntity entity = list.get(i);

            //设置F1数据
            Put put = new Put((entity.getId()+"").getBytes());    //设置行号
            put.addColumn("F1".getBytes(), "id".getBytes(),(entity.getId()+"").getBytes());
            put.addColumn("F1".getBytes(), "task_id".getBytes(), entity.getTask_id().getBytes());
            if (entity.getCreate_time() != null){
                put.addColumn("F1".getBytes(), "create_time".getBytes(), formatter.format(entity.getCreate_time()).getBytes());
            }
            if (entity.getStart_time() != null){
                put.addColumn("F1".getBytes(), "start_time".getBytes(), formatter.format(entity.getStart_time()).getBytes());
            }
            put.addColumn("F1".getBytes(), "out_filter_number".getBytes(), entity.getOut_filter_number().getBytes());
            put.addColumn("F1".getBytes(), "from_run_param".getBytes(), entity.getFrom_run_param().getBytes());

            //设置F2数据
            put.addColumn("F2".getBytes(), "current_retry_times".getBytes(),entity.getCurrent_retry_times().getBytes());
            put.addColumn("F2".getBytes(), "retry_flag".getBytes(), entity.getRetry_flag().getBytes());
            if (entity.getParent_task_id() != null){
                put.addColumn("F2".getBytes(), "parent_task_id".getBytes(), entity.getParent_task_id().getBytes());
            }
            if (entity.getEnd_time() != null){
                put.addColumn("F2".getBytes(), "end_time".getBytes(),  formatter.format(entity.getEnd_time()).getBytes());
            }
            if (entity.getKill_time() != null){
                put.addColumn("F2".getBytes(), "kill_time".getBytes(), formatter.format(entity.getKill_time()).getBytes());
            }
            if (entity.getFailure_time() != null){
                put.addColumn("F2".getBytes(), "failure_time".getBytes(), formatter.format(entity.getFailure_time()).getBytes());
            }
            puts.add(put);
        }
        HBaseOperationDao.putList("BD_TEST", puts);

    }

    public static void main(String[] args) throws SQLException,ClassNotFoundException{
        startMvDate();
    }

}
